package com.lin.manager.rabbitmq.config;

import com.lin.manager.rabbitmq.enums.RabbitEnum;
import com.lin.manager.rabbitmq.properties.CustomRabbitmqProperties;
import com.lin.manager.rabbitmq.service.DefaultRabbitmqMessageRecordService;
import com.lin.manager.rabbitmq.service.RabbitmqMessageRecordService;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.PropertySource;

import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @作者: 林江
 * @创建时间: 2024/4/14
 * @功能:
 */
@EnableConfigurationProperties(RabbitProperties.class)
@PropertySource("classpath:Rabbitmq.properties")
public class RabbitmqConfig {

    @Bean
    @ConfigurationProperties(prefix = "rabbitmq")
    public CustomRabbitmqProperties customRabbitmqProperties() {
        return new CustomRabbitmqProperties();
    }

    @Bean
    public CachingConnectionFactory cachingConnectionFactory(RabbitProperties properties) {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setPort(properties.getPort());
        factory.setHost(properties.getHost());
        factory.setUsername(properties.getUsername());
        factory.setPassword(properties.getPassword());
        factory.setVirtualHost(properties.getVirtualHost());
        // 开始消息的确认机制
        factory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        // 开启消息的返回机制
        factory.setPublisherReturns(true);
        return factory;
    }

    /**
     * 创建监听器容器工厂
     *
     * @return
     */
    @Bean
    public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(CachingConnectionFactory cachingConnectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        /*
         *  手动确认消息
         */
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        factory.setConnectionFactory(cachingConnectionFactory);
        /*
         *  设置一个rabbitmq消费者一次性能处理多少条消息
         */
        factory.setPrefetchCount(2);
        return factory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(CachingConnectionFactory factory,
                                         RabbitmqMessageRecordService logRecordService) {
        RabbitTemplate template = new RabbitTemplate(factory);
        /*
         * 消息投递队列失败之后，才执行回调方法
         */
        template.setMandatory(true);
        // 消息确认的回调
        template.setConfirmCallback((correlationData, ack, cause) -> logRecordService.confirm(correlationData, ack, cause));
        template.setReturnsCallback(returned -> logRecordService.returns(returned));
        return template;
    }

    @Bean
    @ConditionalOnMissingBean
    public RabbitmqMessageRecordService logRecordService() {
        return new DefaultRabbitmqMessageRecordService();
    }

    @Bean
    public RabbitAdmin rabbitAdmin(CachingConnectionFactory factory, CustomRabbitmqProperties properties) {
        RabbitAdmin admin = new RabbitAdmin(factory);

        for (CustomRabbitmqProperties.Rabbitmq rabbitmq : properties.getManager()) {
            // 定义队列
            admin.declareQueue(new Queue(rabbitmq.getQueue(), true, false, false, handler(rabbitmq.getParam())));
            // 定义交换机
            admin.declareExchange(RabbitEnum.getExchange(rabbitmq.getType(), rabbitmq.getExchange()));

            Binding bind = new Binding(
                    rabbitmq.getQueue(),
                    Binding.DestinationType.QUEUE,
                    rabbitmq.getExchange(),
                    rabbitmq.getRoutingkey(),
                    new HashMap<>());
            // 队列和交换机的绑定关系
            admin.declareBinding(bind);
        }

        return admin;
    }

    /**
     * rabbitmq中的字符串数字类型 -> 数值类型
     *
     * @param param
     * @return
     */
    private Map<String, Object> handler(Map<String, Object> param) {
        Map<String, Object> convertMap = new ConcurrentHashMap<>();
        Optional.ofNullable(param).ifPresent(data -> {
            data.forEach((k, v) -> {
                String regx = "\\d+";
                String str = v.toString();
                if (str.matches(regx)) {
                    convertMap.put(k, Integer.parseInt(str));
                } else {
                    convertMap.put(k, v);
                }
            });
        });
        return convertMap;
    }


}
